﻿using Grow.Binary;
using Grow.Domain.EventBus.Cap.DomainEventHandlerInvoker;
using Grow.Domain.EventBus.Cap.DomainEventSend;
using Grow.Domain.EventBus.Cap.Jobs;
using Grow.Domain.EventBus.Cap.Kafka.CreateDomainEventBus;
using Grow.Domain.EventBus.Cap.PublisherDomainEvents;
using Grow.Domain.EventBus.Cap.SubscriberDomainEvents;
using Grow.Domain.EventBus.CreateDomainEventBus;
using Grow.Domain.Repository;
using Grow.Domain.Repository.CreateDomainRepository;
using Grow.Json;
using Grow.Module;
using Grow.Module.Service;
using Grow.Scheduler;
using Grow.Scheduler.CreateScheduler;
using Microsoft.Extensions.DependencyInjection;

namespace Grow.Domain.EventBus.Cap.Kafka
{
    public abstract class DomainEventBusKafKaModuleBase : ModuleBase, IDomainEventBusModule
    {
        public override ModuleLevel Level => ModuleLevel.Framework;


        public override void AddDependModule(IDependModuleBuilder dependModuleBuilder)
        {
            _ = dependModuleBuilder.AddDependModule<IDomainRepositoryModule>();
            _ = dependModuleBuilder.AddDependModule<IBinaryModule>();
            _ = dependModuleBuilder.AddDependModule<IJsonModule>();
            _ = dependModuleBuilder.AddDependModule<ITaskSchedulerModule>();
        }



        public override void AddService(IServiceContainer serviceContainer)
        {
            CreateDomainEventBusBuilder createDomainEventBusBuilder = new(serviceContainer);
            CreateDomainEventBus(createDomainEventBusBuilder);
            serviceContainer.BuilderManager.SetBuilder<ICreateDomainEventBusBuilder>(createDomainEventBusBuilder);

            ICreateDomainRepositoryBuilder createRepositoryBuilder = serviceContainer.BuilderManager.GetRequiredBuilder<ICreateDomainRepositoryBuilder>();
            _ = createRepositoryBuilder.AddDbMap<PublisherDomainEventDbMap>();
            _ = createRepositoryBuilder.AddDbMap<SubscriberDomainEventDbMap>();
            createRepositoryBuilder.AddDomainRepository<IPublisherDomainEventDomainRepository, PublisherDomainEventDomainRepository, PublisherDomainEvent>();
            createRepositoryBuilder.AddDomainRepository<ISubscriberDomainEventDomainRepository, SubscriberDomainEventDomainRepository, SubscriberDomainEvent>();


            ICreateSchedulerBuilder createSchedulerBuilder = serviceContainer.BuilderManager.GetRequiredBuilder<ICreateSchedulerBuilder>();
            _ = createSchedulerBuilder.AddJob<ExpirerPublisherDomainEventJob>();
            _ = createSchedulerBuilder.AddJob<ExpirerSubscriberDomainEventJob>();
            _ = createSchedulerBuilder.AddJob<RetryPublisherDomainEventJob>();
            _ = createSchedulerBuilder.AddJob<RetrySubscriberDomainEventJob>();


            _ = serviceContainer.ServiceCollection.AddTransient<IDomainEventPublisher, DomainEventPublisher>();
            _ = serviceContainer.ServiceCollection.AddTransient<IPublisherDomainEventHandler, PublisherDomainEventHandler>();
            _ = serviceContainer.ServiceCollection.AddTransient<IPublisherDomainEventService, PublisherDomainEventService>();


            _ = serviceContainer.ServiceCollection.AddTransient<IDomainEventSubscriber, DomainEventSubscriber>();
            _ = serviceContainer.ServiceCollection.AddTransient<ISubscriberDomainEventHandler, SubscriberDomainEventHandler>();
            _ = serviceContainer.ServiceCollection.AddTransient<ISubscriberDomainEventService, SubscriberDomainEventService>();

            _ = serviceContainer.ServiceCollection.AddTransient<IDomainEventSend, DomainEventSend>();
            _ = serviceContainer.ServiceCollection.AddTransient<IDomainEventHandlerInvoker, DomainEventHandlerInvoker.DomainEventHandlerInvoker>();

            _ = serviceContainer.ServiceCollection.AddSingleton<IKafkaConnection, KafkaConnection>();
        }



        public override void UseService(IServiceContext serviceContext)
        {
            DomainEventBusOptions eventBusOptions = serviceContext.GetOptions<DomainEventBusOptions>();
            Module.Thread.ICancellationTokenContext cancellationTokenContext = serviceContext.GetCancellationTokenContext();
            IScheduler scheduler = serviceContext.GetRequiredService<IScheduler>();

            scheduler.AddJob<RetryPublisherDomainEventJob>($"{eventBusOptions.EnvironmentName}_RetryPublisherEventData", eventBusOptions.RetryCron, cancellationTokenContext.Token);
            scheduler.AddJob<RetrySubscriberDomainEventJob>($"{eventBusOptions.EnvironmentName}_RetrySubscriberEventData", eventBusOptions.RetryCron, cancellationTokenContext.Token);
            scheduler.AddJob<ExpirerPublisherDomainEventJob>($"{eventBusOptions.EnvironmentName}_ExpirerPublisherEventDataJob", eventBusOptions.ExpirerCron, cancellationTokenContext.Token);
            scheduler.AddJob<ExpirerSubscriberDomainEventJob>($"{eventBusOptions.EnvironmentName}_ExpirerPublisherEventDataJob", eventBusOptions.ExpirerCron, cancellationTokenContext.Token);

            IEnumerable<ISubscriberDomainEvent> subscriberDomainEvents = serviceContext.GetServices<ISubscriberDomainEvent>();
            _ = Parallel.ForEach(subscriberDomainEvents, subscriberDomainEvent =>
            {
                _ = Task.Factory.StartNew(async () =>
                {
                    await subscriberDomainEvent.SubscriberAsync(cancellationTokenContext.Token);
                }, cancellationTokenContext.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
            });
        }

        public abstract void CreateDomainEventBus(ICreateDomainEventBusBuilder createDomainEventBusBuilder);
    }
}
